package com.shujia.flink.state

import java.lang

import org.apache.flink.api.common.functions.RuntimeContext
import org.apache.flink.api.common.state.{ListState, ListStateDescriptor}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

object Demo3ListState {

  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val studentsDS: DataStream[String] = env.socketTextStream("master", 8888)

    /**
      * 实时统计每个班级的平均年龄
      *
      */
    val kvDS: DataStream[(String, Double)] = studentsDS.map(line => {
      val split: Array[String] = line.split(",")

      (split(4), split(2).toDouble)
    })

    kvDS
      .keyBy(_._1)
      .process(new KeyedProcessFunction[String, (String, Double), (String, Double)] {


        var listState: ListState[Double] = _

        //在open方法中初始化状态
        override def open(parameters: Configuration): Unit = {
          val context: RuntimeContext = getRuntimeContext

          val listStatDesc = new ListStateDescriptor[Double]("ages", classOf[Double])

          listState = context.getListState(listStatDesc)

        }

        /**
          * 每一条数据执行一次
          *
          * @param value : 一行数据
          * @param ctx   ： 上下文对象
          * @param out   ： 用于将数据发送到下游
          */
        override def processElement(value: (String, Double), ctx: KeyedProcessFunction[String, (String, Double), (String, Double)]#Context, out: Collector[(String, Double)]): Unit = {

          //每一条数据将年龄保存到集合状态中
          listState.add(value._2)


          //获取每隔班级所有年龄,返回一个java的迭代器
          val ages: lang.Iterable[Double] = listState.get()

          //导入一个隐式转换将java的集合转换成scala的集合
          import scala.collection.JavaConversions._
          val agesList: List[Double] = ages.toList

          /**
            * 注意： 随着数据量不断增加，集合中元素不断增加，对集合进行sum会变得很慢，会导致flink延迟越来越大
            *
            */
          //计算平均年龄
          val avgAge: Double = agesList.sum / agesList.size

          //将数据发送到下游
          out.collect((value._1, avgAge))


        }
      })
      .print()



    env.execute()



  }

}
